package com.reload.core;

import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.io.IOException;
import java.time.Duration;

/**
 * @Date: 2024/4/27 0027 - 15:51
 * @Version: 17
 * @Author: Reloading
 * @Description: 创建序列 与 自定义流中元素的处理规则
 */
public class Sequence {
    public static void main(String[] args) throws IOException {
        generateInteger();
        // generateString();
        // create();
        // dispose();
        System.in.read();
    }

    // 同步环境建议使用
    public static void generateInteger() {
        Flux.generate(
                        // 初始值
                        () -> 0,
                        (state, sink) -> {
                            // 把元素传递出去
                            if (state <= 10) {
                                sink.next("哈哈" + state);
                            } else {
                                // 完成信号
                                sink.complete();
                            }
                            if (state == 7) {
                                sink.error(new Exception("不喜欢7"));
                            }
                            // 返回新的迭代值
                            return state + 1;
                        })
                .doOnError(e -> System.out.println("发生错误：" + e.getMessage()))
                .subscribe(System.out::println);
    }

    // 可结束的
    public static void dispose(){
        Flux<Integer> log = Flux.just(1, 2, 3, 4, 5)
                .map(i -> i + 7)
                .delayElements(Duration.ofSeconds(1))
                .log();
        // 可取消的
        Disposable disposable = log.subscribe(System.out::println);

        new Thread(()->{
            try {
                Thread.sleep(3000);
                disposable.dispose();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }).start();


    }
    public static void generateString() {
        Flux.generate(
                        // 初始值
                        () -> "hello",
                        (state, sink) -> {
                            int i = 0;
                            // 把元素传递出去
                            sink.next("哈哈-" + state);
                            i++;
                            if (state.equals("hello-1-1-1-1-1-1-1-1-1-1-1-1-1-1-1-1-1-1")) {
                                // 完成信号
                                sink.complete();
                            }
                            if (state.equals("hello-1-1-1-1-1-1-1-1-1")) {
                                sink.error(new RuntimeException("不喜欢 -1-1-1-1-1-1-1-1-1"));
                            }
                            // 返回新的初始值
                            return state + "-" + i;
                        })
                .doOnError(e -> System.out.println("发生错误：" + e.getMessage()))
                .subscribe(System.out::println);
    }

    // 多线程环境建议使用
    public static void create() {
        // 拿到异步的通道
        // fluxSink是向Flux流中推送数据的。一个订阅者只执行一次new MyListener(fluxSink);
        // 多订阅者执行多次
        Flux.create(fluxSink -> {
                    MyListener listener = new MyListener(fluxSink);
                    for (int i = 0; i < 10; i++) {
                        listener.online("张 - " + i);
                    }
                })
                .log()
                .subscribe();
    }

    static class MyListener {
        FluxSink<Object> sink;

        public MyListener(FluxSink<Object> sink) {
            this.sink = sink;
        }

        public void online(String userName) {
            System.out.println("用户登录：" + userName);
            sink.next(userName);
        }
    }
}
